package com.browse.config;

import com.browse.service.BrowseRecordService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author 路亚
 * @version 0.1
 * @className RocketmqConfig
 * @description
 * @date 2024/12/26 21:01
 * @since jdk11
 */
@Configuration
public class RocketmqConfig {

    @Value("${luya.mq.name_srv_addr}")
    private String nameSrvAddr;

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer(BrowseRecordService browseRecordService){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("videoBrowseRecord");
        consumer.setNamesrvAddr(nameSrvAddr);
        try {
            consumer.subscribe("videoBrowseRecord", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : list) {
                        String json = new String(messageExt.getBody());
                        browseRecordService.addBrowseRecord(json);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        return consumer;
    }
}
